package cn.doitedu.rtdw.data_sync;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job04_视频信息维表同步hbase {

    public static void main(String[] args) {
        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建业务库中的视频信息表的cdc映射表
        tenv.executeSql("create table cms_video_mysql (  " +
                " id                int               "+
                " ,video_name       string           "+
                " ,video_type       string           "+
                " ,video_album      string           "+
                " ,video_author     string           "+
                " ,video_timelong   bigint           "+
                " ,create_time      timestamp(3)     "+
                " ,update_time      timestamp(3)     "+
                "  ,primary key (id)  not enforced   " +
                ")WITH (                             " +
                "    'connector' = 'mysql-cdc',      " +
                "    'hostname' = 'doitedu',         " +
                "    'port' = '3306',                " +
                "    'username' = 'root',            " +
                "    'password' = 'root',            " +
                "    'database-name' = 'realtimedw', " +
                "    'table-name' = 'cms_video'      " +
                ")");


        // 创建hbase中的视频信息维表的映射表
        tenv.executeSql("create table dim_video_info_hbase(   " +
                "   video_id    int,                   " +
                "   f ROW<video_name string,video_type string, video_album string,video_author string,video_timelong bigint,create_time timestamp(3)>   \n" +
                ") with (                              " +
                "'connector' = 'hbase-2.2',            " +
                "'table-name' = 'dim_video_info',       " +
                "'zookeeper.quorum' = 'doitedu:2181'   " +
                ")");


        tenv.executeSql("insert into dim_video_info_hbase " +
                "select id as video_id,row(video_name,video_type,video_album,video_author,video_timelong,create_time) as f " +
                "from cms_video_mysql");





    }

}
